package com.wdl.networkflow.topN

import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer

import java.text.SimpleDateFormat

object NetWorkFlowTopNPage {

  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    env.setParallelism(1)
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    val inputStream: DataStream[String] =
      env.addSource( new FlinkKafkaConsumer[String](KafkaUtil.topic, new SimpleStringSchema(), KafkaUtil.kafkaConsumerProperties()) )

    /** 装换成样例类类型，指定 timestamp 和 watermark */
    val dataStream: DataStream[ApacheLogEvent] = inputStream.map(data => {
      val dataArray = data.split(" ")
      // 将时间字段转换成时间戳
      val simpleDateFormat = new SimpleDateFormat("dd/MM/yyyy:HH:mm:ss")
      val timestamp = simpleDateFormat.parse(dataArray(3)).getTime

      ApacheLogEvent(dataArray(0), dataArray(1), timestamp, dataArray(5), dataArray(6))
    }).assignTimestampsAndWatermarks(
      new BoundedOutOfOrdernessTimestampExtractor[ApacheLogEvent](Time.seconds(1)) {
        override def extractTimestamp(element: ApacheLogEvent) = {
          element.eventTime
        }
      })

    /** 定义侧输出流 */
    val sideOutputTag = new OutputTag[ApacheLogEvent]("late data")

    val aggStream: DataStream[PageViewCount] = dataStream.keyBy(_.url)
      .timeWindow(Time.minutes(10), Time.seconds(5))
      /** 允许处理数据的最迟时间 */
      .allowedLateness(Time.minutes(1))
      /** 侧输出流 */
      .sideOutputLateData(sideOutputTag)
      .aggregate(new PageCountAgg(), new PageCountWindowResult())

    val sideOutputStream: DataStream[ApacheLogEvent] = aggStream.getSideOutput(sideOutputTag)

    val resultStream: DataStream[String] = aggStream.keyBy(_.windowEnd)
      .process(new TopNHotPage(5))

    resultStream.print("agg")

    env.execute("top n page count job !")
  }

}
